package com.xzx.flink.streamapi.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * bin/zkServer.sh --config conf start
 * bin/kafka-server-start.sh config/server.properties
 * bin/kafka-console-consumer.sh --bootstrap-server 172.25.48.130:29092 --topic click-topic
 * bin/kafka-console-producer.sh --broker-list 172.25.48.130:29092 --topic click-topic
 * @author xinzhixuan
 * @version V1.0.0
 * @date 2021/4/20
 **/
public class SourceFromKafkaV2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // kafka 配置项
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("172.25.48.130:29092")
                .setTopics("click-topic")
                .setGroupId("com/xzx/flink")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 从 kafka 读取数据
        DataStreamSource<String> sourceStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "click Source");

        sourceStream.print();

        env.execute();
    }
}
